package com.bs.service;


import com.bs.domain.ServiceInfo;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsBuilder;
import org.elasticsearch.search.aggregations.metrics.max.InternalMax;
import org.elasticsearch.search.aggregations.metrics.max.MaxBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cglib.beans.BeanCopier;
import org.springframework.cglib.core.Converter;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

@Service
public class SlowSqlTop10Service {

    @Autowired
    ElasticsearchTemplate elasticsearchTemplate;

    private String partition;
    private String uid;
    private String startTime;
    private String endTime;
    private String code;
    private String active;


    private SearchResponse getResponse() {
        String index = "logstash-*-" + active + "-svcsql-*";
        Client transportClient = elasticsearchTemplate.getClient();
        BoolQueryBuilder builder = QueryBuilders.boolQuery();

        if (partition.indexOf("CMS") == -1) {
            builder.filter(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("partition", partition)));
            if ("all".equals(uid)) {
                builder.mustNot(QueryBuilders.termQuery("uid", "batch"))
                    .mustNot(QueryBuilders.termQuery("uid", "xcom"));
            } else {
                builder.filter(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("uid", uid)));

            }
        }

        builder.filter(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("category", "SVCSQL"))
            .must(QueryBuilders.termQuery("priority", "4"))
            .must(QueryBuilders.rangeQuery("@timestamp").from(startTime).to(endTime).timeZone("+08:00")));
//        if(!("all".equals(uid))){
//
//        }
        if (!code.equals("")) {
            index = "logstash-" + code + "-svcsql-*";
        }


        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch(index).setTypes("log");
        TermsBuilder svcAgg = AggregationBuilders.terms("logmsg").field("logmsg.raw")
            .size(10)
            .order(Terms.Order.aggregation("svc_exec_time", false));
        MaxBuilder maxAgg = AggregationBuilders.max("svc_exec_time").field("exec_time");
        TermsBuilder serviceAgg = AggregationBuilders.terms("service").field("svc");
        TermsBuilder partitionAgg = AggregationBuilders.terms("partition").field("partition");
        return searchRequestBuilder
            .setSize(0)
            .setQuery(builder)
            .addAggregation(svcAgg.subAggregation(maxAgg).subAggregation(serviceAgg).subAggregation(partitionAgg))
            .execute().actionGet();
    }

    /**
     * 查询慢SQL Top10
     *
     * @param partition
     * @param uid
     * @param startTime
     * @param endTime
     * @param code
     * @param active
     * @return
     */
    public List<ServiceInfo> slowSqlTop10Resource(String partition, String uid, String startTime, String endTime, String code, String active) {
        this.partition = partition;
        this.startTime = startTime;
        this.endTime = endTime;
        this.uid = uid;
        this.code = code;
        this.active = active;
        List<ServiceInfo> slowServiceInfoList = new ArrayList<>();
        SearchResponse response = getResponse();
        if (response.getHits().totalHits() != 0) {
            StringTerms svcAgg = (StringTerms) response.getAggregations().asMap().get("logmsg");
            Iterator<Terms.Bucket> iterator = svcAgg.getBuckets().iterator();
            while (iterator.hasNext()) {
                Terms.Bucket next = iterator.next();
                String logmsg = next.getKeyAsString();
                InternalMax maxTime = (InternalMax) next.getAggregations().asMap().get("svc_exec_time");
                double value = maxTime.getValue();
                StringTerms partitionMap = (StringTerms) next.getAggregations().asMap().get("partition");
                String partitionKey = (String) partitionMap.getBuckets().get(0).getKey();
                StringTerms serviceMap = (StringTerms) next.getAggregations().asMap().get("service");
                String serviceKey = (String) serviceMap.getBuckets().get(0).getKey();
                ServiceInfo slowServiceInfo = new ServiceInfo(serviceKey, value, logmsg, partitionKey);
                slowServiceInfoList.add(slowServiceInfo);
            }
        }
        return slowServiceInfoList;

    }
}
